package com.xiaochong.tir.trades.scheduled;

import akka.dispatch.OnFailure;
import akka.dispatch.OnSuccess;
import com.xc.logclient.utils.LogTrace;
import com.xiaochong.tir.common.data.mapper.ExchangeInfoMapper;
import com.xiaochong.tir.common.data.po.ExchangeInfo;
import com.xiaochong.tir.common.util.AkkaParamUtils;
import com.xiaochong.tir.common.util.AkkaRouterUtils;
import com.xiaochong.tir.common.util.LogTraceKeysUtils;
import com.xiaochong.tir.common.util.Process;
import com.xiaochong.tir.trades.component.akka.AkkaKActionQuantization;
import com.xiaochong.tir.trades.config.ExecutorConfig;
import com.xiaochong.tir.trades.service.KActionQuantizationService;
import com.xiaochong.tir.trades.thread.KActionQuantizationThread;
import com.xiaochong.tir.trades.util.AkkaUtilTradesParam;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;

/**
 * @Description: 24小时净流入计算
 * @Author: org
 * @CreateDate: 2018/7/2 16:31
 */
@EnableScheduling
@Component
public class KActionQuantizationScheduled {
    private void akkaProcess(String serviceName, Object o){
        LogTrace.beginTrace("akka5分钟执行24小时净流入计算");
        try{
            Process process = AkkaRouterUtils.getProcess(serviceName);
            process.process(o, new OnSuccess() {
                @Override
                public void onSuccess(Object o) {
                    LogTrace.info("akka5分钟执行24小时净流入计算", LogTraceKeysUtils.DATA,"akka执行成功");
                }
            }, new OnFailure() {
                @Override
                public void onFailure(Throwable throwable) {
                    Exception e = new Exception(throwable);
                    e.printStackTrace();
                    LogTrace.error("akka5分钟执行24小时净流入计算"
                            , e, "akka5分钟执行24小时净流入计算");
                }
            },120);
        }catch (Exception e){
            e.printStackTrace();
            LogTrace.error(e.getMessage(), e);
        }

    }

    @Autowired
    private ExchangeInfoMapper exchangeInfoMapper;
    @Autowired
    private KActionQuantizationService kActionQuantizationService;

    java.util.concurrent.Semaphore semaphore = new Semaphore(1, true);


    /**
     *24小时净流入
     */
    @Async(ExecutorConfig.SCHEDULER_EXECUTOR_POOL)
    @Scheduled(cron="0 0/1 * * * ?")
    public void calculateNewHighAndLow() throws InterruptedException {
        LogTrace.beginTrace();
        semaphore.acquire();
        long start = System.currentTimeMillis();
        LogTrace.info("thread5分钟执行24小时净流入计算","start",String.valueOf(start));
        //查询所有交易所，循环调取akka
        ExchangeInfo exchangeInfo=new ExchangeInfo();
        exchangeInfo.setEnableFlag( true );
        exchangeInfo.setDeleteFlag( false );
        exchangeInfo.setKlineStatus( true );

        List<ExchangeInfo> infoList=exchangeInfoMapper.selectByEntityList( exchangeInfo );
        int i =1;
        CountDownLatch countDownLatch = new CountDownLatch(infoList.size());
        for(ExchangeInfo exchange : infoList){
            /*Map<String,ExchangeInfo> akkaExchageMap = new HashMap<>();
            akkaExchageMap.put("exchange",exchange);
            String serverName = AkkaUtilTradesParam.AKKA_TRADES_K_ACTION_QUANTIZATION+"_"+i;
            AkkaRouterUtils.registerAsync(serverName, AkkaKActionQuantization.class);
            this.akkaProcess(serverName,akkaExchageMap);
            i ++;*/
            KActionQuantizationThread kThread = new KActionQuantizationThread(exchange.getId(),countDownLatch);
            new Thread(kThread).start();
        }
        countDownLatch.await();
        semaphore.release();
        LogTrace.info("thread5分钟执行24小时净流入计算","end",String.valueOf(System.currentTimeMillis()-start));
    }

    /**
     * 计算24小时最高最低价格、成交量K线数据,昨日收盘价格
     */
    @Async(ExecutorConfig.SCHEDULER_EXECUTOR_POOL)
    @Scheduled(fixedRate = 1000*60*60*365)
    public void calculateKActionHighAndLowInit(){
        LogTrace.beginTrace();
        long start = System.currentTimeMillis();
        LogTrace.info("K线-24小时最高最低价格、成交量Init","start",String.valueOf(start));
        kActionQuantizationService.calculateKActionHighAndLowVolume();
        LogTrace.info("K线-24小时最高最低价格、成交量Init","end",String.valueOf(System.currentTimeMillis()-start));
    }
    /**
     * 计算24小时最高最低价格、成交量K线数据,昨日收盘价格
     */
    @Async(ExecutorConfig.SCHEDULER_EXECUTOR_POOL)
    @Scheduled(cron="0 0/1 0/1 * * ?")
    public void calculateKActionHighAndLow() {
        LogTrace.beginTrace();
        long start = System.currentTimeMillis();
        LogTrace.info("K线-24小时最高最低价格、成交量","start",String.valueOf(start));
        kActionQuantizationService.calculateKActionHighAndLowVolume();
        LogTrace.info("K线-24小时最高最低价格、成交量","end",String.valueOf(System.currentTimeMillis()-start));
    }

    /**
     * 从redis获取昨日收盘价格
     */
    @Async(ExecutorConfig.SCHEDULER_EXECUTOR_POOL)
    @Scheduled(fixedRate = 1000*60*60*365)
    public void calculateYesterdayClosePriceInit(){
        LogTrace.beginTrace();
        long start = System.currentTimeMillis();
        LogTrace.info("从redis获取昨日收盘价格Init","start",String.valueOf(start));
        kActionQuantizationService.calculateYesterdayClosePriceInit();
        LogTrace.info("从redis获取昨日收盘价格Init","end",String.valueOf(System.currentTimeMillis()-start));
    }
    /**
     * 从redis获取昨日收盘价格
     */
    @Async(ExecutorConfig.SCHEDULER_EXECUTOR_POOL)
    @Scheduled(cron="0 0/20 1 * * ?")
    public void calculateYesterdayClosePrice() {
        LogTrace.beginTrace();
        long start = System.currentTimeMillis();
        LogTrace.info("从redis获取昨日收盘价格","start",String.valueOf(start));
        kActionQuantizationService.calculateYesterdayClosePriceInit();
        LogTrace.info("从redis获取昨日收盘价格","end",String.valueOf(System.currentTimeMillis()-start));
    }
}
